package com.johnguo.carbondata_demo;

import java.io.Serializable;

import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.CarbonContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.SaveMode;

public class LoadData {
	public static void main(String... args) {
		SparkConf conf = new SparkConf().setAppName("insertData");
		SparkContext sc = new SparkContext(conf);
		JavaSparkContext jsc = new JavaSparkContext(sc);
		CarbonContext cc = new CarbonContext(sc, "hdfs://master:9000/usr/cardata");
		JavaRDD<String> rdd1 = jsc.textFile("/test/*");
		JavaRDD<Sale> rdd2 = rdd1.map(new Function<String, Sale>() {

			@Override
			public Sale call(String val) throws Exception {
				Sale sale = new Sale();
				String vals[] = val.split("\\|");
				sale.setName(vals[0]);
				sale.setArea(vals[1]);
				sale.setBrand(vals[2]);
				sale.setSale_pay(Double.valueOf(vals[3]));
				sale.setPay_num(Integer.valueOf(vals[4]));
				sale.setTotal_pay(Double.valueOf(vals[5]));
				sale.setDate(vals[6]);
				return sale;
			}
		});
		DataFrame df = cc.createDataFrame(rdd2, Sale.class);
		df.write().format("org.apache.spark.sql.CarbonSource").option("tableName", "sale")
		.mode(SaveMode.Append).save();
	}

	public static class Sale implements Serializable{
		private String name;
		private String area;
		private String brand;
		private Double sale_pay;
		private Integer pay_num;
		private Double total_pay;
		private String date;

		public String getName() {
			return name;
		}

		public void setName(String name) {
			this.name = name;
		}

		public String getArea() {
			return area;
		}

		public void setArea(String area) {
			this.area = area;
		}

		public String getBrand() {
			return brand;
		}

		public void setBrand(String brand) {
			this.brand = brand;
		}

		public Double getSale_pay() {
			return sale_pay;
		}

		public void setSale_pay(Double sale_pay) {
			this.sale_pay = sale_pay;
		}

		public Integer getPay_num() {
			return pay_num;
		}

		public void setPay_num(Integer pay_num) {
			this.pay_num = pay_num;
		}

		public Double getTotal_pay() {
			return total_pay;
		}

		public void setTotal_pay(Double total_pay) {
			this.total_pay = total_pay;
		}

		public String getDate() {
			return date;
		}

		public void setDate(String date) {
			this.date = date;
		}

	}

}
